package simpledb.execution;

import simpledb.common.DbException;
import simpledb.common.Type;
import simpledb.storage.IntField;
import simpledb.storage.StringField;
import simpledb.storage.Tuple;
import simpledb.storage.TupleDesc;
import simpledb.transaction.TransactionAbortedException;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

import java.io.IOException;
import java.util.*;

/**
 * Knows how to compute some aggregate over a set of StringFields.
 */
public class StringAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    /**
     * Aggregate constructor
     * @param gbfield the 0-based index of the group-by field in the tuple, or NO_GROUPING if there is no grouping
     * @param gbfieldtype the type of the group by field (e.g., Type.INT_TYPE), or null if there is no grouping
     * @param afield the 0-based index of the aggregate field in the tuple
     * @param what aggregation operator to use -- only supports COUNT
     * @throws IllegalArgumentException if what != COUNT
     */
    private int gbfield;
    private Type gbfieldtype;
    private int afield;
    private Op op;
    Map<String, List<String>> gbIdxToAggList;

    public StringAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here
        this.gbfield=gbfield;
        this.afield=afield;
        this.op=what;
        this.gbfieldtype=gbfieldtype;
        gbIdxToAggList=new HashMap<>();
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the constructor
     * @param tup the Tuple containing an aggregate field and a group-by field
     */
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        String key="NoGrouping";
        if(!(gbfield==Aggregator.NO_GROUPING)&&(gbfieldtype==Type.INT_TYPE)){
            IntField field= (IntField) tup.getField(gbfield);
            key=Integer.toString(field.getValue());
        }
        else if(!(gbfield==Aggregator.NO_GROUPING)){
            StringField stringField=(StringField) tup.getField(gbfield);
            key=stringField.getValue();
        }
        if(!gbIdxToAggList.containsKey(key)){
            gbIdxToAggList.put(key,new ArrayList<>());
        }
        StringField stringField= (StringField) tup.getField(afield);
        gbIdxToAggList.get(key).add(stringField.getValue());

    }

    /**
     * Create a OpIterator over group aggregate results.
     *
     * @return a OpIterator whose tuples are the pair (groupVal,
     *   aggregateVal) if using group, or a single (aggregateVal) if no
     *   grouping. The aggregateVal is determined by the type of
     *   aggregate specified in the constructor.
     */
    public OpIterator iterator() {
        // some code goes here

       return new OpIterator() {
           private Iterator iterator;
           private int calculate(List<String>stringList){
               assert !stringList.isEmpty() : "expect non empty list";
               if(op==Op.COUNT){
                   return stringList.size();
               }
               throw new NotImplementedException();
           }

           @Override
           public void open() throws DbException, TransactionAbortedException {
               iterator=gbIdxToAggList.entrySet().iterator();
           }

           @Override
           public boolean hasNext() throws DbException, TransactionAbortedException {
               if(iterator==null)
                   return false;
               return iterator.hasNext();
           }

           @Override
           public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException {
               if (iterator == null)
                   throw new IllegalStateException("Operator not yet open");

               if (!iterator.hasNext()) {
                   throw new NoSuchElementException();
               }
               //拿列表。
               Map.Entry<String,List<String>>stringListEntry= (Map.Entry<String, List<String>>) iterator.next();
               List<String>list=stringListEntry.getValue();
               String key=stringListEntry.getKey();
               IntField res=new IntField(calculate(list));


               TupleDesc tupleDesc=getTupleDesc();
               Tuple tuple=new Tuple(tupleDesc);

               if(gbfield==Aggregator.NO_GROUPING){
                  tuple.setField(0,res);
               }
               else if(gbfieldtype==Type.INT_TYPE){
                   IntField intField=new IntField(Integer.parseInt(key));
                   tuple.setField(0,intField);
                   tuple.setField(1,res);
               }
               else{
                   StringField stringField=new StringField(key,key.length());
                   tuple.setField(0,stringField);
                   tuple.setField(1,res);
               }
               return tuple;

           }

           @Override
           public void rewind() throws DbException, TransactionAbortedException {
               close();
               open();

           }

           @Override
           public TupleDesc getTupleDesc() {
              if(gbfield==Aggregator.NO_GROUPING){
                  return new TupleDesc(new Type[]{Type.INT_TYPE});
              }
              else{
                  return new TupleDesc(new Type[]{gbfieldtype,Type.INT_TYPE});
              }
           }

           @Override
           public void close() {
               iterator=null;
           }
       };
    }

}
